package sunyu.tools.hbase;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * hbase原生java操作工具类
 *
 * @author 孙宇
 */
@Component
public class HbaseTools {

    public static final String aggregateImplementationCoprocessor = "org.apache.hadoop.hbase.coprocessor.AggregateImplementation";//统计协处理器
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    private Connection connection;
    private AggregationClient aggregationClient;
    private LongColumnInterpreter longColumnInterpreter = new LongColumnInterpreter();
    @Value("${hbase.zookeeper.quorum}")
    private String quorum;
    @Value("${hbase.zookeeper.znode.parent}")
    private String znode;
    @Value("${hbase.encoding}")
    private String encoding;

    @PostConstruct
    public void init() {
        logger.info("初始化Hbase开始");
        logger.info("hbase.zookeeper.quorum：{}", quorum);
        logger.info("zookeeper.znode.parent：{}", znode);
        Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", quorum);
        config.set("zookeeper.znode.parent", znode);
        config.set("hbase.encoding", encoding);

        try {
            connection = ConnectionFactory.createConnection(config);
        } catch (IOException e) {
            logger.error(ExceptionUtils.getStackTrace(e));
        }

        aggregationClient = new AggregationClient(config);
        logger.info("初始化Hbase结束");
    }

    @PreDestroy
    public void destroy() {
        logger.info("销毁Hbase连接池开始");
        if (!connection.isClosed()) {
            try {
                connection.close();
            } catch (IOException e) {
                logger.error(ExceptionUtils.getStackTrace(e));
            }
        }
        try {
            aggregationClient.close();
        } catch (IOException e) {
            logger.error(ExceptionUtils.getStackTrace(e));
        }
        logger.info("销毁Hbase连接池结束");
    }

    /**
     * 获得Admin
     *
     * @param connection
     *
     * @return
     */
    private Admin getAdmin(Connection connection) {
        try {
            return connection.getAdmin();
        } catch (IOException e) {
            logger.error(ExceptionUtils.getStackTrace(e));
        }
        return null;
    }

    /**
     * 创建表
     *
     * @param tableName
     * @param familyNames
     */
    public void createTable(String tableName,
                            String[] familyNames) {
        try (Admin admin = getAdmin(connection)) {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
            hTableDescriptor.addCoprocessor(aggregateImplementationCoprocessor);
            for (String familyName : familyNames) {
                HColumnDescriptor family = new HColumnDescriptor(familyName);
                hTableDescriptor.addFamily(family);
            }
            admin.createTable(hTableDescriptor);
        } catch (IOException e) {
            logger.error(ExceptionUtils.getStackTrace(e));
        }
    }

    /**
     * 添加协处理器
     *
     * @param tableName
     * @param className
     */
    public void addCoprocessor(String tableName,
                               String className) {
        HTableDescriptor desc = null;
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            desc = table.getTableDescriptor();
        } catch (IOException e) {
            logger.error(ExceptionUtils.getStackTrace(e));
        }
        if (!desc.getCoprocessors().contains(className)) {
            Admin admin = getAdmin(connection);
            try {
                admin.disableTable(TableName.valueOf(tableName));
                HTableDescriptor hTableDescriptor = admin.getTableDescriptor(TableName.valueOf(tableName));
                hTableDescriptor.addCoprocessor(className);
                admin.modifyTable(TableName.valueOf(tableName), hTableDescriptor);
            } catch (IOException e) {
                logger.error(ExceptionUtils.getStackTrace(e));
            } finally {
                try {
                    admin.enableTable(TableName.valueOf(tableName));
                } catch (IOException e) {
                    logger.error(ExceptionUtils.getStackTrace(e));
                }
                try {
                    admin.close();
                } catch (IOException e) {
                    logger.error(ExceptionUtils.getStackTrace(e));
                }
            }
        }
    }

    /**
     * 删除表
     *
     * @param tableName
     */
    public void deleteTable(String tableName) {
        try (Admin admin = getAdmin(connection)) {
            admin.disableTable(TableName.valueOf(tableName));
            admin.deleteTable(TableName.valueOf(tableName));
        } catch (IOException e) {
            logger.error(ExceptionUtils.getStackTrace(e));
        }
    }

    /**
     * 获得table描述
     *
     * @param tableName
     *
     * @return
     */
    public HTableDescriptor tableDescriptor(String tableName) {
        HTableDescriptor desc = null;
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            desc = table.getTableDescriptor();
        } catch (IOException e) {
            logger.error(ExceptionUtils.getStackTrace(e));
        }
        return desc;
    }

    /**
     * 插入一行
     *
     * @param tableName
     * @param put
     */
    public void put(String tableName,
                    Put put) {
        put(tableName, Arrays.asList(put));
    }

    /**
     * 插入多行
     *
     * @param tableName
     * @param puts
     */
    public void put(String tableName,
                    List<Put> puts) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            table.put(puts);
        } catch (IOException e) {
            logger.error(ExceptionUtils.getStackTrace(e));
        }
    }

    /**
     * 扫描
     *
     * @param tableName
     *
     * @return
     */
    public List<Result> scan(String tableName) {
        return scan(tableName, new Scan());
    }

    /**
     * 扫描
     *
     * @param tableName
     * @param pageSize
     *
     * @return
     */
    public List<Result> scan(String tableName,
                             Integer pageSize) {
        return scan(tableName, new Scan(), pageSize);
    }

    /**
     * 扫描
     *
     * @param tableName
     * @param scan
     *
     * @return
     */
    public List<Result> scan(String tableName,
                             Scan scan) {
        return scan(tableName, scan, null);
    }

    /**
     * 扫描
     *
     * @param tableName
     * @param scan
     * @param pageSize
     *
     * @return
     */
    public List<Result> scan(String tableName,
                             Scan scan,
                             Integer pageSize) {
        List<Result> results = new ArrayList<>();
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            try (ResultScanner resultScanner = table.getScanner(scan)) {
                if (pageSize != null) {
                    Result[] rs = resultScanner.next(pageSize);
                    for (Result result : rs) {
                        results.add(result);
                    }
                } else {
                    for (Result result : resultScanner) {
                        results.add(result);
                    }
                }
            } catch (Exception ex) {
                logger.error(ExceptionUtils.getStackTrace(ex));
            }
        } catch (IOException e) {
            logger.error(ExceptionUtils.getStackTrace(e));
        }
        return results;
    }

    /**
     * 删除一行
     *
     * @param tableName
     * @param del
     */
    public void delete(String tableName,
                       Delete del) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            table.delete(del);
        } catch (IOException e) {
            logger.error(ExceptionUtils.getStackTrace(e));
        }
    }

    /**
     * 获得一行
     *
     * @param tableName
     * @param g
     *
     * @return
     */
    public Result get(String tableName,
                      Get g) {
        Result[] result = get(tableName, Arrays.asList(g));
        if (result != null && result.length > 0) {
            return result[0];
        }
        return null;
    }

    /**
     * 获得多行
     *
     * @param tableName
     * @param gets
     *
     * @return
     */
    public Result[] get(String tableName,
                        List<Get> gets) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            return table.get(gets);
        } catch (IOException e) {
            logger.error(ExceptionUtils.getStackTrace(e));
        }
        return null;
    }

    /**
     * 统计行数，依赖协处理器
     *
     * @param tableName
     * @param scan
     *
     * @return
     */
    public long rowCount(String tableName,
                         Scan scan) {
        try {
            return aggregationClient.rowCount(TableName.valueOf(tableName), longColumnInterpreter, scan);
        } catch (Throwable e) {
            logger.error(ExceptionUtils.getStackTrace(e));
        }
        return -1;
    }

}
